package com.qunar.toku.socket;

import static org.jboss.netty.channel.Channels.pipeline;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.concurrent.Executors;

import org.jboss.netty.bootstrap.ClientBootstrap;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;
import org.jboss.netty.handler.codec.serialization.ClassResolvers;
import org.jboss.netty.handler.codec.serialization.ObjectDecoder;
import org.jboss.netty.handler.codec.serialization.ObjectEncoder;
import org.jboss.netty.logging.InternalLogger;
import org.jboss.netty.logging.InternalLoggerFactory;

import com.qunar.toku.configuration.SocketPoolProfile;
import com.qunar.toku.handler.ChannelClientHandler;
import com.qunar.toku.proxy.RequestProtocol;

/**
 * CSocket
 * 
 * @author zhangwan
 * 
 */
public class CSocket {
	private final InternalLogger logger = 
			InternalLoggerFactory.getInstance(CSocket.class);
	private Channel channel;
	private boolean _connecting = false;
	private boolean _inPool = false;
	
	public CSocket(final SocketPoolProfile profile, InetSocketAddress endPoint, final ChannelClientHandler h) {
		ClientBootstrap bootstrap = new ClientBootstrap(new NioClientSocketChannelFactory(Executors.newCachedThreadPool(),
				Executors.newCachedThreadPool()));

		bootstrap.setPipelineFactory(new ChannelPipelineFactory() {

			@Override
			public ChannelPipeline getPipeline() throws Exception {
				ChannelPipeline pipeline = pipeline();
				pipeline.addLast("encode", new ObjectEncoder(profile.getMaxPakageSize()));
				pipeline.addLast("decode", new ObjectDecoder(profile.getMaxPakageSize(), ClassResolvers.weakCachingConcurrentResolver(null)));
				pipeline.addLast("handler", h);
				return pipeline;
			}
		});

		bootstrap.setOption("receiveBufferSize", profile.getRecvBufferSize());
		bootstrap.setOption("sendBufferSize", profile.getSendBufferSize());
		bootstrap.setOption("tcpNoDelay", profile.isNagle()); // 关闭Nagle算法
		bootstrap.setOption("keepAlive", profile.isKeepalive());
		ChannelFuture future = bootstrap.connect(endPoint);
		channel = future.awaitUninterruptibly().getChannel();
		_connecting = true;
		
		logger.info("MaxPakageSize:" + profile.getMaxPakageSize());
		logger.info("SendBufferSize:" + profile.getSendBufferSize());
		logger.info("RecvBufferSize:" + profile.getRecvBufferSize());
        logger.info("create a new connection :" + this.toString());
	}

	public Channel getChannel() {
		return channel;
	}

	public void send(RequestProtocol protocol) {
		channel.write(protocol);
	}

	protected void disconnect() throws IOException {
		if (channel != null) {
			channel.close();
		}
		
		_connecting = false;
	}
	
	public boolean connecting() {
        return _connecting;
    }
	
	protected boolean inPool() {
        return _inPool;
    }

	protected void setInPool(boolean inPool) {
        _inPool = inPool;
    }
	
    @Override
    public String toString() {
        try {
            return channel.toString() + " _connecting:" + _connecting + " _inPool:" + _inPool; 
        } catch (Throwable ex) {
            return "Socket[error:" + ex.getMessage() + "]";
        }
    }

}
